package com.yanggu.zookeeper;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.junit.Before;
import org.junit.Test;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class CuratorDemo4 {

    private CuratorFramework client;

    /**
     * 创建重试策略和客户端
     */
    @Before
    public void createClient() {
        //重试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        //创建客户端
        client = CuratorFrameworkFactory.newClient("localhost:2181", retryPolicy);
        //启动客户端
        client.start();
    }

    /**
     * 简单使用
     *
     * @throws Exception
     */
    @Test
    public void simpleUse() throws Exception {
        String result = client.create().forPath("/testCurator", "test1".getBytes(StandardCharsets.UTF_8));
        System.out.println(result);
    }

    /**
     * 创建节点和为节点设置值
     *
     * @throws Exception
     */
    @Test
    public void createAndSetValue() throws Exception {
        //新增节点
        String forPath = client.create()
                //withMode 设置节点类型。具体类型在CreateMode枚举中定义。默认为持久化节点
                .withMode(CreateMode.EPHEMERAL)
                //forPath 定义节点路径, 可以设置值或者不设置
                .forPath("/testCurator2", "test2".getBytes(StandardCharsets.UTF_8));
        System.out.println(forPath);
    }

    /**
     * 递归创建节点，如果不存在就创建
     *
     * @throws Exception
     */
    @Test
    public void creatingParentsIfNeeded() throws Exception {
        client.create()
                .creatingParentsIfNeeded()
                .withProtection()
                .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                .forPath("/test4/test5", "haha".getBytes(StandardCharsets.UTF_8));
    }

    /**
     * 删除节点
     *
     * @throws Exception
     */
    @Test
    public void delete() throws Exception {
        System.out.println(client.delete().forPath("/test1"));
    }

    /**
     * 删除节点并递归删除其子节点
     *
     * @throws Exception
     */
    @Test
    public void deletingChildrenIfNeeded() throws Exception {
        System.out.println(client.delete().deletingChildrenIfNeeded().forPath("/test4"));
    }

    /**
     * 指定版本进行删除。
     * 如果版本不存在，则删除异常org.apache.zookeeper.KeeperException$BadVersionException: KeeperErrorCode = BadVersion for /test2
     *
     * @throws Exception
     */
    @Test
    public void deleteWithVersion() throws Exception {
        System.out.println(client.delete().withVersion(2).forPath("/test2"));
    }

    /**
     * 获取数据
     * 普通查询和包含状态查询
     *
     * @throws Exception
     */
    @Test
    public void getData() throws Exception {
        System.out.println(new String(client.getData().forPath("/test3")));
        Stat stat = new Stat();
        System.out.println(new String(client.getData().storingStatIn(stat).forPath("/test3")));
        System.out.println(stat);
    }

    /**
     * 获取指定路径下的子节点
     *
     * @throws Exception
     */
    @Test
    public void getChildren() throws Exception {
        List<String> list = client.getChildren().forPath("/");
        System.out.println(list);
    }

    /**
     * 更新数据。普通更新
     */
    @Test
    public void updateData() throws Exception {
        Stat stat = client.setData().forPath("/test3", "新数据".getBytes(StandardCharsets.UTF_8));
        System.out.println(stat);
    }

    /**
     * 更新数据，指定版本
     * 如果版本不对，抛出异常
     * org.apache.zookeeper.KeeperException$BadVersionException: KeeperErrorCode = BadVersion for /test3
     */
    @Test
    public void updateDataWithVersion() throws Exception {
        Stat stat = client.setData().withVersion(2).forPath("/test3", "新数据2".getBytes(StandardCharsets.UTF_8));
        System.out.println(stat);
    }

    /**
     * 数据节点变化的监听器
     *
     * @throws Exception
     */
    @Test
    public void nodeDataUpdateWatcher() throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        NodeCache nodeCache = new NodeCache(client, "/test3");
        nodeCache.start();
        nodeCache.getListenable().addListener(
                () -> System.out.println("Node data is changed, new data " + nodeCache.getCurrentData().getData()),
                pool);
        TimeUnit.MILLISECONDS.sleep(Long.MAX_VALUE);
    }

    /**
     * 监听子节点的变化
     *
     * @throws Exception
     */
    @Test
    public void childChangeWatcher() throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/test3", true);
        pathChildrenCache.start();
        pathChildrenCache.getListenable().addListener((
                (client1, event) -> {
                    switch (event.getType()) {
                        case CHILD_ADDED:
                            System.out.println("CHILD_ADDED: " + event.getData().getPath());
                            break;
                        case CHILD_REMOVED:
                            System.out.println("CHILD_REMOVED: " + event.getData().getPath());
                            break;
                        case CHILD_UPDATED:
                            System.out.println("CHILD_UPDATED: " + event.getData().getPath());
                            break;
                        default:
                            break;
                    }
                }
        ), pool);
        TimeUnit.MILLISECONDS.sleep(Long.MAX_VALUE);
    }

    /**
     * 分布式锁测试类
     */
    @Test
    public void distrlockTest() throws Exception {
        Thread t1 = new Thread(
                () -> doWithLock(client),
                "t1"
        );
        t1.start();

        Thread t2 = new Thread(
                () -> doWithLock(client),
                "t2"
        );
        t2.start();
        TimeUnit.MILLISECONDS.sleep(Long.MAX_VALUE);
    }

    private static void doWithLock(CuratorFramework client) {
        InterProcessMutex lock = new InterProcessMutex(client, "/lock");
        while (true) {
            try {
                if (lock.acquire(10 * 1000, TimeUnit.SECONDS)) {
                    System.out.println(Thread.currentThread().getName() + " hold lock");
                    Thread.sleep(5000L);
                    System.out.println(Thread.currentThread().getName() + " release lock");
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    lock.release();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * leader选取
     */
    @Test
    public void testLeaderElection() throws Exception {

        LeaderSelectorListener listener = new LeaderSelectorListenerAdapter() {

            @Override
            public void takeLeadership(CuratorFramework client) throws Exception {

                System.out.println(Thread.currentThread().getName() + " take leadership!");
                // takeLeadership() method should only return when leadership is being relinquished.

                Thread.sleep(5000L);
                System.out.println(Thread.currentThread().getName() + " relinquish leadership!");
            }
        };

        new Thread(() -> registerListener(listener), "t1").start();
        new Thread(() -> registerListener(listener), "t2").start();
        new Thread(() -> registerListener(listener), "t3").start();

        TimeUnit.MILLISECONDS.sleep(Long.MAX_VALUE);
    }

    private void registerListener(LeaderSelectorListener listener) {
        LeaderSelector leaderSelector = new LeaderSelector(client, "/leaderElection", listener);
        leaderSelector.autoRequeue();
        leaderSelector.start();
    }

}
